use async_trait::async_trait;
use fluviox_shared::{Result, FluvioXError, models::DataRecord};
use crate::traits::{SourceConnector, ConnectorConfig, ConnectorStatus, ConnectorMetrics};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use reqwest::Client;
use std::collections::HashMap;
use uuid::Uuid;
use chrono::Utc;
use tracing::{info, error, warn};

/// HTTP数据源配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HttpSourceConfig {
    /// API端点URL
    pub url: String,
    /// HTTP方法
    pub method: String,
    /// 请求头
    pub headers: HashMap<String, String>,
    /// 请求体（用于POST/PUT请求）
    pub body: Option<String>,
    /// 轮询间隔（秒）
    pub poll_interval: u64,
    /// 认证配置
    pub auth: Option<AuthConfig>,
    /// 分页配置
    pub pagination: Option<PaginationConfig>,
    /// 数据提取路径（JSONPath）
    pub data_path: Option<String>,
    /// 超时时间（秒）
    pub timeout: Option<u64>,
}

/// 认证配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum AuthConfig {
    /// Bearer Token认证
    Bearer { token: String },
    /// Basic认证
    Basic { username: String, password: String },
    /// API Key认证
    ApiKey { key: String, value: String, location: ApiKeyLocation },
}

/// API Key位置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ApiKeyLocation {
    Header,
    Query,
}

/// 分页配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PaginationConfig {
    /// 分页类型
    pub page_type: PaginationType,
    /// 页面大小参数名
    pub page_size_param: String,
    /// 页面大小
    pub page_size: u32,
    /// 页码参数名（用于页码分页）
    pub page_param: Option<String>,
    /// 偏移量参数名（用于偏移量分页）
    pub offset_param: Option<String>,
    /// 游标参数名（用于游标分页）
    pub cursor_param: Option<String>,
}

/// 分页类型
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum PaginationType {
    /// 页码分页
    Page,
    /// 偏移量分页
    Offset,
    /// 游标分页
    Cursor,
    /// 无分页
    None,
}

impl ConnectorConfig for HttpSourceConfig {
    fn validate(&self) -> Result<()> {
        // 验证URL格式
        if self.url.is_empty() {
            return Err(FluvioXError::Validation("URL不能为空".to_string()));
        }
        
        // 验证HTTP方法
        match self.method.to_uppercase().as_str() {
            "GET" | "POST" | "PUT" | "DELETE" | "PATCH" => {},
            _ => return Err(FluvioXError::Validation("不支持的HTTP方法".to_string())),
        }
        
        // 验证轮询间隔
        if self.poll_interval == 0 {
            return Err(FluvioXError::Validation("轮询间隔必须大于0".to_string()));
        }
        
        Ok(())
    }
    
    fn to_json(&self) -> Result<Value> {
        Ok(serde_json::to_value(self)?)
    }
    
    fn from_json(value: Value) -> Result<Self> {
        Ok(serde_json::from_value(value)?)
    }
}

/// HTTP数据源连接器
pub struct HttpSourceConnector {
    config: Option<HttpSourceConfig>,
    client: Client,
    status: ConnectorStatus,
    metrics: ConnectorMetrics,
    last_cursor: Option<String>,
    current_page: u32,
    current_offset: u32,
}

impl HttpSourceConnector {
    pub fn new() -> Self {
        Self {
            config: None,
            client: Client::new(),
            status: ConnectorStatus::Uninitialized,
            metrics: ConnectorMetrics::default(),
            last_cursor: None,
            current_page: 1,
            current_offset: 0,
        }
    }
    
    /// 构建HTTP请求
    async fn build_request(&self, config: &HttpSourceConfig) -> Result<reqwest::Request> {
        let mut url = config.url.clone();
        
        // 处理分页参数
        if let Some(pagination) = &config.pagination {
            url = self.add_pagination_params(url, pagination)?;
        }
        
        let mut request_builder = match config.method.to_uppercase().as_str() {
            "GET" => self.client.get(&url),
            "POST" => self.client.post(&url),
            "PUT" => self.client.put(&url),
            "DELETE" => self.client.delete(&url),
            "PATCH" => self.client.patch(&url),
            _ => return Err(FluvioXError::Validation("不支持的HTTP方法".to_string())),
        };
        
        // 添加请求头
        for (key, value) in &config.headers {
            request_builder = request_builder.header(key, value);
        }
        
        // 添加认证
        if let Some(auth) = &config.auth {
            request_builder = self.add_auth(request_builder, auth)?;
        }
        
        // 添加请求体
        if let Some(body) = &config.body {
            request_builder = request_builder.body(body.clone());
        }
        
        // 设置超时
        if let Some(timeout) = config.timeout {
            request_builder = request_builder.timeout(std::time::Duration::from_secs(timeout));
        }
        
        request_builder.build()
            .map_err(|e| FluvioXError::Http(e.to_string()))
    }
    
    /// 添加分页参数
    fn add_pagination_params(&self, mut url: String, pagination: &PaginationConfig) -> Result<String> {
        let separator = if url.contains('?') { "&" } else { "?" };
        
        match pagination.page_type {
            PaginationType::Page => {
                if let Some(page_param) = &pagination.page_param {
                    url.push_str(&format!("{}{}={}", separator, page_param, self.current_page));
                    url.push_str(&format!("&{}={}", pagination.page_size_param, pagination.page_size));
                }
            },
            PaginationType::Offset => {
                if let Some(offset_param) = &pagination.offset_param {
                    url.push_str(&format!("{}{}={}", separator, offset_param, self.current_offset));
                    url.push_str(&format!("&{}={}", pagination.page_size_param, pagination.page_size));
                }
            },
            PaginationType::Cursor => {
                if let Some(cursor_param) = &pagination.cursor_param {
                    if let Some(cursor) = &self.last_cursor {
                        url.push_str(&format!("{}{}={}", separator, cursor_param, cursor));
                    }
                    url.push_str(&format!("&{}={}", pagination.page_size_param, pagination.page_size));
                }
            },
            PaginationType::None => {},
        }
        
        Ok(url)
    }
    
    /// 添加认证信息
    fn add_auth(&self, mut builder: reqwest::RequestBuilder, auth: &AuthConfig) -> Result<reqwest::RequestBuilder> {
        match auth {
            AuthConfig::Bearer { token } => {
                builder = builder.bearer_auth(token);
            },
            AuthConfig::Basic { username, password } => {
                builder = builder.basic_auth(username, Some(password));
            },
            AuthConfig::ApiKey { key, value, location } => {
                match location {
                    ApiKeyLocation::Header => {
                        builder = builder.header(key, value);
                    },
                    ApiKeyLocation::Query => {
                        // 这里需要修改URL，但为了简化，我们先添加到header
                        builder = builder.header(key, value);
                    },
                }
            },
        }
        
        Ok(builder)
    }
    
    /// 提取数据记录
    fn extract_records(&self, response_body: &str, config: &HttpSourceConfig) -> Result<Vec<DataRecord>> {
        let json_value: Value = serde_json::from_str(response_body)?;
        let mut records = Vec::new();
        
        // 根据数据路径提取数据
        let data = if let Some(path) = &config.data_path {
            // 简单的JSONPath实现（只支持基本的路径）
            self.extract_by_path(&json_value, path)?
        } else {
            json_value
        };
        
        // 将数据转换为记录
        match data {
            Value::Array(items) => {
                for item in items {
                    records.push(DataRecord {
                        id: Uuid::new_v4(),
                        topic: "http-source".to_string(), // 这里应该从配置中获取
                        partition: 0,
                        offset: 0, // 这里应该是实际的偏移量
                        timestamp: Utc::now(),
                        data: item,
                        metadata: HashMap::new(),
                    });
                }
            },
            _ => {
                // 单个对象也作为一条记录
                records.push(DataRecord {
                    id: Uuid::new_v4(),
                    topic: "http-source".to_string(),
                    partition: 0,
                    offset: 0,
                    timestamp: Utc::now(),
                    data,
                    metadata: HashMap::new(),
                });
            }
        }
        
        Ok(records)
    }
    
    /// 简单的JSONPath提取实现
    fn extract_by_path(&self, value: &Value, path: &str) -> Result<Value> {
        let parts: Vec<&str> = path.split('.').collect();
        let mut current = value;
        
        for part in parts {
            if part.is_empty() {
                continue;
            }
            
            current = current.get(part)
                .ok_or_else(|| FluvioXError::Internal(format!("路径 '{}' 不存在", part)))?;
        }
        
        Ok(current.clone())
    }
}

#[async_trait]
impl SourceConnector for HttpSourceConnector {
    type Config = HttpSourceConfig;
    
    fn name(&self) -> &str {
        "http-source"
    }
    
    fn version(&self) -> &str {
        "1.0.0"
    }
    
    fn description(&self) -> &str {
        "HTTP/REST API数据源连接器，支持轮询、分页和认证"
    }
    
    async fn initialize(&mut self, config: Self::Config) -> Result<()> {
        info!("初始化HTTP数据源连接器");
        
        // 验证配置
        config.validate()?;
        
        self.config = Some(config);
        self.status = ConnectorStatus::Initialized;
        
        info!("HTTP数据源连接器初始化完成");
        Ok(())
    }
    
    async fn test_connection(&self) -> Result<bool> {
        info!("测试HTTP连接");
        
        let config = self.config.as_ref()
            .ok_or_else(|| FluvioXError::Internal("连接器未初始化".to_string()))?;
        
        match self.build_request(config).await {
            Ok(request) => {
                match self.client.execute(request).await {
                    Ok(response) => {
                        let success = response.status().is_success();
                        if success {
                            info!("HTTP连接测试成功");
                        } else {
                            warn!("HTTP连接测试失败，状态码: {}", response.status());
                        }
                        Ok(success)
                    },
                    Err(e) => {
                        error!("HTTP连接测试失败: {}", e);
                        Ok(false)
                    }
                }
            },
            Err(e) => {
                error!("构建HTTP请求失败: {}", e);
                Ok(false)
            }
        }
    }
    
    async fn start(&mut self) -> Result<()> {
        info!("启动HTTP数据源连接器");
        self.status = ConnectorStatus::Running;
        Ok(())
    }
    
    async fn stop(&mut self) -> Result<()> {
        info!("停止HTTP数据源连接器");
        self.status = ConnectorStatus::Stopped;
        Ok(())
    }
    
    async fn read_records(&mut self, _batch_size: usize) -> Result<Vec<DataRecord>> {
        let config = self.config.as_ref()
            .ok_or_else(|| FluvioXError::Internal("连接器未初始化".to_string()))?;
        
        let request = self.build_request(config).await?;
        
        match self.client.execute(request).await {
            Ok(response) => {
                if response.status().is_success() {
                    let body = response.text().await
                        .map_err(|e| FluvioXError::Http(e.to_string()))?;
                    let records = self.extract_records(&body, config)?;
                    
                    // 更新指标
                    self.metrics.records_processed += records.len() as u64;
                    self.metrics.last_activity = Some(Utc::now());
                    
                    // 更新分页状态
                    if let Some(pagination) = &config.pagination {
                        match pagination.page_type {
                            PaginationType::Page => {
                                self.current_page += 1;
                            },
                            PaginationType::Offset => {
                                self.current_offset += pagination.page_size;
                            },
                            _ => {},
                        }
                    }
                    
                    Ok(records)
                } else {
                    self.metrics.errors_count += 1;
                    Err(FluvioXError::Internal(format!("HTTP请求失败，状态码: {}", response.status())))
                }
            },
            Err(e) => {
                self.metrics.errors_count += 1;
                error!("HTTP请求执行失败: {}", e);
                Err(FluvioXError::Internal(format!("HTTP请求执行失败: {}", e)))
            }
        }
    }
    
    fn status(&self) -> ConnectorStatus {
        self.status.clone()
    }
    
    fn metrics(&self) -> ConnectorMetrics {
        self.metrics.clone()
    }
}
